package com.tang.window;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SessionWindowTimeGapExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 时间窗口Demo
 * ====================
 * 触发器、移除器：现成的几个窗口，都有默认的实现，一般不需要自己定义
 * ====================
 * 以时间类型的滚动窗口为例，分析原理：
 * 1、窗口什么时候触发输出？
 * 时间进展 >= 窗口最大时间戳（end -1ms）
 * -
 * 2、窗口是怎么划分的？
 * 
 *
 * @author tang
 * @since 2023/6/11 15:56
 */
public class TimeWindowDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KeyedStream<WaterSensor, String> keyedStream = env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .keyBy(WaterSensor::getId);

        // 滚动窗口
        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // 滑动窗口
//        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(20), Time.seconds(10)));
        // 会话窗口, 固定时间
//        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));
        // 会话窗口，动态时间
//        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(ProcessingTimeSessionWindows.withDynamicGap(WaterSensor::getTs));
//        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream.window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<WaterSensor>() {
//            @Override
//            public long extract(WaterSensor element) {
//                return element.getTs();
//            }
//        }));

        window.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<WaterSensor, String, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                // 上下文可以拿到window对象，还有其他东西：侧输出流 等等
                long startTs = context.window().getStart();
                long endTs = context.window().getEnd();
                String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
                String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");

                long count = elements.spliterator().estimateSize();

                out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
            }
        }).print();

        env.execute();
    }

}
